package com.lee.mq.mqserver.core;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: lishuo
 * Date: 2023-10-27
 * Time: 18:35
 */

import com.lee.mq.common.Consumer;
import com.lee.mq.common.ConsumerEnv;
import com.lee.mq.common.MqException;
import com.lee.mq.mqserver.VirtualHost;
import org.apache.ibatis.executor.Executor;

import java.io.IOException;
import java.util.concurrent.*;

/**
 * 实现消费的核心逻辑
 */
public class ConsumerManager {
    // 具体消费消息的思路如下：
    // 1.创建一个线程池取处理consumer参数中的业务逻辑
    // 2.创建一个存放令牌的阻塞队列（存放不为空的队列名）
    // 3.创建一个扫描线程，持续扫描令牌队列，进行消费

    // 创建一个VirtualHost实例，后续会使用到里面的方法
    private VirtualHost virtualHost = null;
    // 创建线程池
    private ExecutorService workPool= Executors.newFixedThreadPool(4);
    // 创建存放令牌的阻塞队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    // 创建扫描线程
    private Thread scannerThread = null;

    // 构造方法中写入初始化时的一些逻辑
    public ConsumerManager(VirtualHost virtualHost) {
        this.virtualHost = virtualHost;

        // 初始化扫描线程（扫描不为空的队列）
        scannerThread = new Thread(()->{
            // 持续扫描令牌队列
            while (true) {
                try {
                    String queueName = tokenQueue.take();
                    MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 取出令牌后发现，队列不存在");
                    }
                    // 给订阅者推送消息：处理consumer参数中的具体逻辑
                    // 这里的令牌队列存储的是队列名，每有一个队列中插入了新消息，就会将队列名插入到里面，所以每次只消费一条消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }

                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        // 这里将扫描线程设置为后台线程，并启动线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    // 发送消息的时候将对列名添加到令牌队列中
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    // 给指定队列添加订阅者
    public void addConsumerEnv(String consumerTag, String queueName, boolean autoAsk, Consumer consumer) throws MqException {
        // 获取并判断队列是否为空
        queueName = virtualHost.getVirtualHostName() + queueName;
        MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列为空 queueName = "+queueName);
        }
        // 封装ConsumerEnv对象
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAsk,consumer);

        synchronized (queue) {
            // 将ConsumerEnv对象插入到指定队列中
            queue.addConsumerEvn(consumerEnv);
            // 获取队列中的消息数量
            int n = virtualHost.getMemoryDataCenter().getMessageCount(queueName);
            // 如果队列中已经存在消息，则消费掉（1、这个步骤是必要的，因为可能是生产者先上线，生产了一些消息后，消费者才上线，这样的话，上面的扫描线程
            // 拿到队列名后由于没有订阅者也无能为力，就直接把令牌中的队列名取走啥也不干。因此这里需要进行保底策略，如果出现上面的情况，就在这里进行
            // 消费。2、如果是消费者先上线，也就是先有订阅，在插入消息，那么这里的消息数量就为空，这段逻辑也就不在起作用了）
            for (int i = 0; i < n; i ++) {
                // 每调用一次消费一条消息
                consumeMessage(queue);
            }
        }
    }

    // 给订阅者推送消息，根据consumer规则进行消费
    private void consumeMessage(MSGQueue queue) {
        // 1.使用轮询的方式从队列中取出一个订阅者
        ConsumerEnv luckyConsumer = queue.chooseConsumerEvn();
        if (luckyConsumer == null) {
            // 不存在订阅者，暂不消费
            return;
        }
        // 2.从内存中取消息
        Message message = virtualHost.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            // 不存在消息，暂不处理
            return;
        }
        // 3.使用线程池将消息带入到回调方法上
        workPool.submit(()->{
            try {
                // 执行回调之前先将消息添加到待应答容器中
                virtualHost.getMemoryDataCenter().addMessageWaitAsk(queue.getName(),message);
                // 真正执行consumer回调函数
                luckyConsumer.getConsumer().handleDelivery(luckyConsumer.getConsumerTag(),
                        message.getBasicProperties(), message.getBody());
                // 如果是自动应答，直接把消息删除即可
                // 如果是手动应答，这里就先不处理，等待后续消费者调用basicAck实现
                if (luckyConsumer.isAutoAck()) {
                    // 删除硬盘数据
                    if (message.getDeliverMode() == 2) {
                        virtualHost.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    // 删除内存数据
                    virtualHost.getMemoryDataCenter().removeMessageWaitAsk(queue.getName(),message.getMessageId());
                    virtualHost.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息消费成功 messageId = "+message.getMessageId()
                            +",queueName = "+queue.getName());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}
